package com.lry.crawlers.listener;

import com.lry.crawlers.config.RabbitMQConfig;
import com.lry.crawlers.video.entity.VideoUpdate;
import com.lry.crawlers.video.entity.VideoVariable;
import com.lry.crawlers.video.pipeline.VideoPipeline;
import com.lry.crawlers.video.processor.VideoUpdateProcessor;
import com.lry.crawlers.video.service.VideoUpdateService;
import com.lry.crawlers.video.service.VideoVariableService;
import com.lry.crawlers.information.entity.Information;
import com.lry.crawlers.information.service.InformationService;
import com.lry.crawlers.message.service.MessageService;
import com.lry.crawlers.util.ProcessVariable;
import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import us.codecraft.webmagic.Spider;

import java.util.*;

@Component
public class videoUpdateListener {
    @Autowired
    VideoUpdateService videoUpdateService;
    @Autowired
    VideoPipeline videoPipeline;
    @Autowired
    VideoVariableService videoVariableService;
    @Autowired
    InformationService informationService;
    @Autowired
    MessageService messageService;


    //监听video_queue队列
    @RabbitListener(queues = RabbitMQConfig.VUPDATEEQUEUE_NAME)
    public void ListenerUpdateQueue(Message message, Channel channel) throws Exception {
        String uuid =UUID.randomUUID().toString();
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        String str = new String(message.getBody());
        int i = Integer.valueOf(str.split(",")[0]);
        int m = Integer.valueOf(str.split(",")[1]);
        videoVariableService.update();
        List<VideoUpdate> videoUpdates = videoUpdateService.getVideoUpdates();
        videoUpdates = averageAssign(videoUpdates, m).get(i);
        try {
            Thread.sleep(2000);
            informationService.setInformation(new Information(uuid,"视频","视频更新， 参与更新节点数："+m+"， 当前节点临时编号："+(i)+" ，更新数量："+videoUpdates.size(),"进行中",new Date(),null));

            Map<String, Integer> map = new HashMap<>();
            for (VideoUpdate vu : videoUpdates) {
                VideoVariable vv = ProcessVariable.getVVariable(vu.getVideoVariableId());
                String url = vv.getProtocol() + "://" + vv.getUrl() + vu.getVideoUrl();
                map.put(url.trim(), vu.getChapterNum());
            }
            Spider.create(new VideoUpdateProcessor(map))
                    //初始化url
                    .addUrl("https://www.baidu.com")
                    //执行爬虫
                    .thread(10)
                    .addPipeline(videoPipeline)
                    .run();
            channel.basicAck(deliveryTag, false);
            informationService.updateInformation(new Information(uuid,"视频","视频更新， 参与更新节点数："+m+"， 当前节点临时编号："+(i)+" ，更新数量："+videoUpdates.size(),"完成",null,new Date()));

        }catch (Exception e){
            channel.basicNack(deliveryTag, false, false);
            informationService.updateInformation(new Information(uuid,"视频","视频更新， 参与更新节点数："+m+"， 当前节点临时编号："+(i)+" ，更新数量："+videoUpdates.size(),"错误",null,new Date()));
            messageService.setMessage(new com.lry.crawlers.message.entity.Message("视频更新出现错误！异常："+e,new Date()));

        }

        }



    public static <T> List<List<T>> averageAssign(List<T> source, int n) {
        List<List<T>> result = new ArrayList<List<T>>();
        int remainder = source.size() % n;  //(先计算出余数)
        int number = source.size() / n;  //然后是商
        int offset = 0;//偏移量
        for (int i = 0; i < n; i++) {
            List<T> value = null;
            if (remainder > 0) {
                value = source.subList(i * number + offset, (i + 1) * number + offset + 1);
                remainder--;
                offset++;
            } else {
                value = source.subList(i * number + offset, (i + 1) * number + offset);
            }
            result.add(value);
        }
        return result;
    }


    }

